package priv.lg.foreachWriter

import org.apache.commons.dbutils.handlers.ArrayHandler
import org.apache.commons.dbutils.{DbUtils, QueryRunner}
import org.apache.spark.sql.ForeachWriter
import priv.lg.bean.BusInfo
import priv.lg.foreachWriter.MysqlWriter.getConnection

import java.sql.{Connection, DriverManager}

class MysqlWriter extends ForeachWriter[BusInfo]{

  private var conn: Connection = _
  private var runner : QueryRunner = _

  /** 开启连接 */
  override def open(partitionId: Long, epochId: Long): Boolean = {
    conn = getConnection()
    runner = new QueryRunner()
    true
  }

  /** 处理 */
  override def process(value: BusInfo): Unit = {
    val lglat: String = value.lglat
    val deployNum: String = value.deployNum
    val sql =
      s"""
         |insert into lg_table(lglat,deployNum)
         |values('$lglat', '$deployNum')
         |""".stripMargin
    runner.update(conn, sql)
  }

  /** 释放连接 */
  override def close(errorOrNull: Throwable): Unit = {
    conn.close()
  }
}

object MysqlWriter {
  val JDBC_DRIVER = "com.mysql.jdbc.Driver"
  val DB_URL = "jdbc:mysql://hadoop3:3306/tmp"
  val USER = "root"
  val PASS = "12345678"

  def getConnection(): Connection = {
    DbUtils.loadDriver(JDBC_DRIVER)
    DriverManager.getConnection(DB_URL, USER, PASS)
  }

}
